package com.jh.cloud.stream.server;

import com.jh.cloud.stream.server.stream.SimpleMessageConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;

@SpringBootApplication
@EnableBinding(SimpleMessageConsumer.class)
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Autowired
    private SimpleMessageConsumer simpleMessageConsumer;

    /**
     * Spring Cloud Stream 接口驱动
     */
    @PostConstruct
    public void consumeMessage(){

        // 获取channel
        SubscribableChannel channel = simpleMessageConsumer.input();

        // 消费消息
        channel.subscribe(message -> {
            System.out.println("--- Spring Cloud Stream 接口驱动编程模型 ---");
            MessageHeaders headers = message.getHeaders();
            String encoding = (String) headers.get("charset-encoding");
            byte[] payload = (byte[]) message.getPayload();
            try {
                System.out.println("message（String）：" + new String(payload, StringUtils.isEmpty(encoding)? "UTF-8" : encoding));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        });

    }

    /**
     * Spring Cloud Stream 注解驱动
     */
    @StreamListener("jh2019")
    public void onMessageAnnotation(String message){
        System.out.println("--- Spring Cloud Stream 注解驱动编程模型 ---");
        System.out.println("message（String）：" + message);
    }

    /**
     * Spring Integration 注解编程模型
     */
    @ServiceActivator(inputChannel = "jh2019")
    public void onMessageActivator(String message) {
        System.out.println("--- Spring Integration 注解驱动编程模型 ---");
        System.out.println("message（String）：" + message);
    }

    @StreamListener("test007")
    public void onStreamMQBinder(byte[] message){
        System.out.println("--- Spring Cloud Stream Binder Test ---");
        System.out.println("message（String）：" + new String(message));
    }

}
